package dht.table;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Vector;

import dht.IClientRouteTable;
import dht.IDataServerRouteTable;
import dht.IMetaServerRouteTable;
import dht.entity.GlobalVariables;
import dht.entity.NodeInfo;

public class MetaServerRouteTableImpl extends DataServerRouteTableImpl
		implements IMetaServerRouteTable {

	private static final long serialVersionUID = 1L;

	protected int replicaLevel;

	protected HashMap<Integer, Integer> vnodeId2Load;

	protected HashSet<String> lockedServerIds;

	public MetaServerRouteTableImpl(NodeInfo masterInfo) {
		super(masterInfo);
		this.replicaLevel = GlobalVariables.replicaLevel;
		this.vnodeId2Load = new HashMap<Integer, Integer>();
		for (int i = 0; i < ringLength; ++i) {
			vnodeId2Load.put(i, 1);
		}
		this.lockedServerIds = new HashSet<String>();
		load(GlobalVariables.fsImage);
	}

	@Override
	public synchronized IClientRouteTable getClientRouteTable() {
		return (IClientRouteTable) ((ClientRouteTableImpl) this).tableClone();
	}

	@Override
	public synchronized IDataServerRouteTable getDataServerRouteTable() {
		return (IDataServerRouteTable) ((DataServerRouteTableImpl) this)
				.tableClone();
	}

	@Override
	public synchronized void join(NodeInfo nodeInfo) {
		if (!registedServer.containsKey(nodeInfo.getServerId())) {
			registedServer.put(nodeInfo.getServerId(), nodeInfo);
			System.out.println(nodeInfo.getServerId());
			// if (registedServer.size() == 1) {
			// for (HashSet<String> servers : vnodeId2ServerIds.values()) {
			// servers.add(nodeInfo.getServerId());
			// }
			// }
		} else {
			// this server leave DHT and rejoin, do recover (if possible) here
		}
	}

	@Override
	public synchronized void leave(NodeInfo nodeInfo) {
		registedServer.remove(nodeInfo.getServerId());
		// for (HashSet<String> servers : vnodeId2ServerIds.values()) {
		// servers.remove(nodeInfo.getServerId());
		// }
	}

	@Override
	public synchronized void tableUpdate(String moveFromServerId,
			String moveToServerId, Vector<Integer> movementVnodeIds) {
		for (Integer vnodeId : movementVnodeIds) {
			vnodeId2ServerIds.get(vnodeId).add(moveToServerId);
			if (vnodeId2ServerIds.get(vnodeId).size() > replicaLevel) {
				vnodeId2ServerIds.get(vnodeId).remove(moveFromServerId);
			}
		}
		save(GlobalVariables.fsImage);
	}

	@Override
	public synchronized void endLoadBalancing(NodeInfo moveFromServerInfo,
			NodeInfo moveToServerInfo) {
		lockedServerIds.remove(moveFromServerInfo.getServerId());
		lockedServerIds.remove(moveToServerInfo.getServerId());
	}

	@Override
	public synchronized boolean startLoadBalancing(NodeInfo moveFromServerInfo,
			NodeInfo moveToServerInfo, Vector<Integer> movementVnodeIds) {
		if (moveFromServerInfo == null || moveToServerInfo == null
				|| movementVnodeIds == null) {
			return false;
		}
		if (registedServer.size() < 2) {
			return false;
		}

		HashMap<String, Integer> serverId2Load = new HashMap<String, Integer>();
		HashMap<String, HashSet<Integer>> serverId2VnodeIds = new HashMap<String, HashSet<Integer>>();
		for (String serverId : registedServer.keySet()) {
			serverId2Load.put(serverId, 0);
			serverId2VnodeIds.put(serverId, new HashSet<Integer>());
		}
		for (Integer vnodeId : vnodeId2ServerIds.keySet()) {
			HashSet<String> serverIds = vnodeId2ServerIds.get(vnodeId);
			for (String serverId : serverIds) {
				if (serverId2Load.containsKey(serverId)) {
					serverId2Load.put(serverId, serverId2Load.get(serverId)
							+ vnodeId2Load.get(vnodeId));
					serverId2VnodeIds.get(serverId).add(vnodeId);
				}
			}
		}
		for (String serverId : serverId2Load.keySet()) {
			System.out.println("serverId: " + serverId + " load: "
					+ serverId2Load.get(serverId));
		}
		for (String serverId : lockedServerIds) {
			System.out.println("lockedServerId: " + serverId);
		}

		String maxLoadServerId = null, minLoadServerId = null;
		int maxLoad = Integer.MIN_VALUE, minLoad = Integer.MAX_VALUE;
		for (String serverId : serverId2Load.keySet()) {
			if (serverId2Load.get(serverId) < minLoad
					&& !lockedServerIds.contains(serverId)) {
				minLoad = serverId2Load.get(serverId);
				minLoadServerId = serverId;
			}
		}
		for (String serverId : serverId2Load.keySet()) {
			if (serverId2Load.get(serverId) > maxLoad
					&& !serverId.equals(minLoadServerId)
					&& !lockedServerIds.contains(serverId)) {
				maxLoad = serverId2Load.get(serverId);
				maxLoadServerId = serverId;
			}
		}

		if (maxLoadServerId == null || minLoadServerId == null) {
			System.out.println("maxLoadServerId: " + maxLoadServerId
					+ " minLoadServerId: " + minLoadServerId
					+ " no need to do balance return");
			return false;
		}

		if ((maxLoad - minLoad) < GlobalVariables.balanceThreshold) {
			System.out.println("minLoad: " + minLoad + " maxLoad: " + maxLoad
					+ " no need to do balance return");
			return false;
		}

		NodeInfo nodeInfo = registedServer.get(maxLoadServerId);
		moveFromServerInfo.setIp(nodeInfo.getIp());
		moveFromServerInfo.setPort(nodeInfo.getPort());
		nodeInfo = registedServer.get(minLoadServerId);
		moveToServerInfo.setIp(nodeInfo.getIp());
		moveToServerInfo.setPort(nodeInfo.getPort());

		HashSet<Integer> minLoadServerVnodeIds = serverId2VnodeIds
				.get(minLoadServerId);
		HashSet<Integer> maxLoadServerVnodeIds = serverId2VnodeIds
				.get(maxLoadServerId);

		int moveLoad = 0;
		// do replication
		// for (Integer vnodeId : maxLoadServerVnodeIds) {
		// if (!minLoadServerVnodeIds.contains(vnodeId)
		// && vnodeId2ServerIds.get(vnodeId).size() < replicaLevel) {
		// moveLoad += vnodeId2Load.get(vnodeId);
		// movementVnodeIds.add(vnodeId);
		// }
		// }

		// do balancing
		int upperMoveLoad = (maxLoad - minLoad + 1) / 2;
		for (Integer vnodeId : maxLoadServerVnodeIds) {
			if (!minLoadServerVnodeIds.contains(vnodeId)
					&& !movementVnodeIds.contains(vnodeId)
					&& moveLoad + vnodeId2Load.get(vnodeId) <= upperMoveLoad) {
				moveLoad += vnodeId2Load.get(vnodeId);
				movementVnodeIds.add(vnodeId);
				if (movementVnodeIds.size() >= GlobalVariables.moveVnodeNumber) {
					break;
				}
			}
		}
		if (movementVnodeIds.size() == 0) {
			return false;
		}
		lockedServerIds.add(moveFromServerInfo.getServerId());
		lockedServerIds.add(moveToServerInfo.getServerId());
		return true;
	}

	@SuppressWarnings("unchecked")
	@Override
	public synchronized void load(String fileName) {
		FileInputStream fis = null;
		ObjectInputStream ois = null;
		try {
			fis = new FileInputStream(fileName);
			ois = new ObjectInputStream(fis);
		} catch (IOException e) {
			System.out.println("File not found: " + fileName);
			return;
		}
		HashMap<Integer, HashSet<String>> obj = null;
		try {
			obj = (HashMap<Integer, HashSet<String>>) ois.readObject();
			ois.close();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			e.printStackTrace();
		}
		if (obj != null && obj.size() == ringLength) {
			vnodeId2ServerIds = obj;
		}
	}

	@Override
	public synchronized void save(String fileName) {
		try {
			FileOutputStream fos = new FileOutputStream(fileName);
			ObjectOutputStream oos = new ObjectOutputStream(fos);
			oos.writeObject(vnodeId2ServerIds);
			oos.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}
